package com.example.demo.service;

import com.alibaba.fastjson.JSON;
import com.example.demo.helper.DingDingHelper;
import com.example.demo.model.Notify;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.util.Optional;

/**
 * 消息通知消费者
 * 1. 监听Kafka的notify主题
 * 2. 反序列化消息
 * 3. 根据消息的平台调用不同的helper
 */
@Component
public class NotifyConsumer {
    @Autowired
    DingDingHelper dingDingHelper;

    /**
     * 监听Kafka的notify主题
     * @param record 消息记录
     */
    @KafkaListener(topics = "notify")
    public void linsten(ConsumerRecord<?,?> record) {
        Optional<?> kafkaMessage = Optional.ofNullable(record.value());
        if (kafkaMessage.isPresent()) {
            Object message = kafkaMessage.get();
            Notify notify= JSON.parseObject(message.toString(), Notify.class);//对Kafka消息序列化
            //根据不同平台调用不同helper
            switch (notify.getPlatform()){
                case DINGDING://通知是钉钉平台的
                    dingDingHelper.sendNotify(notify);
                    break;
                default:
                    break;
            }
        }

    }
}
